package com.supermarket.processors;

import com.supermarket.config.MySQLConfig;
import com.supermarket.utils.JDBCUtil;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.streaming.api.java.JavaDStream;

import java.sql.Connection;
import java.sql.PreparedStatement;

public class CategoryCountProcessor {
    //根据商品类别统计净变化量（上架减去下架）
    //从 MySQL 加载商品维度表 product_dim。
    //与实时数据进行 Join，按类别聚合计算净变化量。
    //插入或更新 MySQL 表 rt_category_count
    public static void process(JavaDStream<String> jsonDStream, SparkSession spark) {
        // 从MySQL加载商品维度数据
        Dataset<Row> productDim = spark.read()
                .format("jdbc")
                .option("url", MySQLConfig.URL)
                .option("dbtable", "product_dim")
                .option("user", MySQLConfig.USER)
                .option("password", MySQLConfig.PASSWORD)
                .load();

        productDim.createOrReplaceTempView("product_dim");

        jsonDStream.foreachRDD(rdd -> {
            if (rdd.isEmpty()) return;

            Dataset<Row> df = spark.read().json(spark.createDataset(rdd.rdd(), Encoders.STRING()));
            df.createOrReplaceTempView("supermarket_data");

            Dataset<Row> result = spark.sql(
                    "SELECT " +
                            "  p.category, " +
                            "  SUM(CASE WHEN s.u_d = 'U' THEN s.u_d_num ELSE -s.u_d_num END) as count " +
                            "FROM supermarket_data s " +
                            "JOIN product_dim p ON s.gid = p.gid " +
                            "GROUP BY p.category"
            );

            result.foreach(row -> {
                try (Connection conn = JDBCUtil.getConnection();
                     PreparedStatement stmt = conn.prepareStatement(
                             "INSERT INTO rt_category_count (category, count) " +
                                     "VALUES (?, ?) " +
                                     "ON DUPLICATE KEY UPDATE " +
                                     "count = count + VALUES(count)")) {

                    stmt.setString(1, row.getString(0));
                    stmt.setLong(2, row.getLong(1));
                    stmt.executeUpdate();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        });
    }
}